package com.zyx.flink14demo.stream.connector;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.pulsar.source.PulsarSource;
import org.apache.flink.connector.pulsar.source.PulsarSourceOptions;
import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.pulsar.client.api.SubscriptionType;

/**
 * @author Yaxi.Zhang
 * @since 2021/10/9 16:48
 * desc: flink的pulsar连接器使用
 * @see <a href="https://nightlies.apache.org/flink/flink-docs-release-1.14/zh/docs/connectors/datastream/pulsar/">flink的pulsar连接器</a>
 */
public class PulsarSourceDemo {

    private static final String SERVICE_URL = "pulsar://bigdata022:6650,bigdata023:6650,bigdata024:6650";
    private static final String ADMIN_URL = "http://bigdata022:8080,bigdata023:8080,bigdata024:8080";
    public static final String TOPICS = "persistent://manning/chapter04/filtered-feed";

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        PulsarSource<String> source = PulsarSource.builder()
                // Pulsar 数据消费的地址
                .setServiceUrl(SERVICE_URL)
                // Pulsar HTTP 管理地址
                .setAdminUrl(ADMIN_URL)
                .setStartCursor(StartCursor.earliest())
                // 需要消费的 topic 或者是 topic 下面的分区
                .setTopics(TOPICS)
                // 解码 Pulsar 消息的反序列化器
                .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(new SimpleStringSchema()))
                // Pulsar 订阅名称
                .setSubscriptionName("my-subshare")
                // 订阅模式, 这里使用 共享模式
                .setSubscriptionType(SubscriptionType.Exclusive)
                // Pulsar的broker.conf中transactionCoordinatorEnabled 默认为false, 由于共享模式下 Flink 需要向Pulsar提交确认, 需要在Pulsar上开启事务
                // 这里为了测试, 将PULSAR_ENABLE_AUTO_ACKNOWLEDGE_MESSAGE设置为true, 此时消息从 Pulsar 消费后会被立刻置为已读, 可能带来数据的不一致
                .setConfig(PulsarSourceOptions.PULSAR_ENABLE_AUTO_ACKNOWLEDGE_MESSAGE, true)
                .build();

        DataStreamSource<String> pulsarDataStream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar Source");
        pulsarDataStream.print();

        env.execute("pulsar-source-demo");

    }

}
